package com.wl.es7.study.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HeaderElement;
import org.apache.http.HeaderElementIterator;
import org.apache.http.HttpHost;
import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.message.BasicHeaderElementIterator;
import org.apache.http.nio.reactor.IOReactorException;
import org.apache.http.protocol.HTTP;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.AbstractFactoryBean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.Assert;

@Slf4j
@Configuration
public class CustomElasticSearchConfig extends AbstractFactoryBean<RestHighLevelClient> {


    // 15分钟 (毫秒)
    private static final long KEEP_ALIVE_MS = 15 * 60 * 1000;

    @Value("${spring.elasticsearch.rest.uris}")
    private String uris;

    @Override
    public Class<?> getObjectType() {
        return RestHighLevelClient.class;
    }

    @Override
    protected RestHighLevelClient createInstance() throws Exception {
        return buildClient();
    }


    /***
     * 配置连接池
     * @return
     * @throws IOReactorException
     */
    private RestHighLevelClient buildClient() throws IOReactorException {

        if (StringUtils.isBlank(uris)) {
            throw new RuntimeException("elasticsearch uri is unset to properties file");
        }

        String[] nodes = uris.split(",");
        HttpHost[] httpHosts = new HttpHost[nodes.length];
        for (int x = 0; x < nodes.length; x++) {
            String[] uris = nodes[x].split(":");
            HttpHost httpHost = new HttpHost(uris[0], Integer.parseInt(uris[1]), "http");
            httpHosts[x] = httpHost;
        }

        IOReactorConfig ioReactorConfig = IOReactorConfig.custom()
                .setIoThreadCount(2)
                .setConnectTimeout(10)
                .setRcvBufSize(512)
                .setSoKeepAlive(true)
                .build();
        PoolingNHttpClientConnectionManager connManager = new PoolingNHttpClientConnectionManager(new DefaultConnectingIOReactor(ioReactorConfig));
        connManager.setMaxTotal(100);
        connManager.setDefaultMaxPerRoute(100);

        // CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
        // credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(username, password));

        RestClientBuilder restClientBuilder = RestClient.builder(httpHosts)
                .setHttpClientConfigCallback(httpClientBuilder -> {
                    // 连接保持
                    httpClientBuilder.setKeepAliveStrategy((response, context) -> {
                        final HeaderElementIterator it = new BasicHeaderElementIterator(
                                response.headerIterator(HTTP.CONN_KEEP_ALIVE));
                        while (it.hasNext()) {
                            final HeaderElement he = it.nextElement();
                            final String param = he.getName();
                            final String value = he.getValue();
                            if (value != null && param.equalsIgnoreCase("timeout")) {
                                try {
                                    return Long.parseLong(value) * 1000;
                                } catch (final NumberFormatException ignore) {
                                }
                            }
                        }
                        return KEEP_ALIVE_MS;
                    });
                    // 最大连接数
                    httpClientBuilder.setMaxConnTotal(100);
                    // 最大路由连接数
                    httpClientBuilder.setMaxConnPerRoute(100);
                    // httpClientBuilder.disableAuthCaching();
                    // httpClientBuilder.setDefaultCredentialsProvider();
                    httpClientBuilder.setConnectionManager(connManager);
                    return httpClientBuilder;
                })
                .setRequestConfigCallback(requestConfigBuilder -> {
                    // 连接目标url的连接超时时间（毫秒）
                    requestConfigBuilder.setConnectTimeout(5000);
                    // 请求获取数据的超时时间(即响应时间)，单位毫秒
                    requestConfigBuilder.setSocketTimeout(40000);
                    // 从connect Manager(连接池)获取Connection 超时时间，单位毫秒
                    requestConfigBuilder.setConnectionRequestTimeout(1000);
                    return requestConfigBuilder;
                });

        return new RestHighLevelClient(restClientBuilder);
    }

    private RestHighLevelClient buildClient2() throws IOReactorException {


        String[] nodes = uris.split(",");
        HttpHost[] httpHosts = new HttpHost[nodes.length];
        for (int x = 0; x < nodes.length; x++) {
            String[] uris = nodes[x].split(":");
            HttpHost httpHost = new HttpHost(uris[0], Integer.parseInt(uris[1]), "http");
            httpHosts[x] = httpHost;
        }

        final IOReactorConfig ioReactorConfig = IOReactorConfig
                .custom()
                .setConnectTimeout(10)
                .setRcvBufSize(5)
                .setSoKeepAlive(true)
                .build();
        final PoolingNHttpClientConnectionManager connManager = new PoolingNHttpClientConnectionManager(new
                DefaultConnectingIOReactor(ioReactorConfig));
        connManager.setMaxTotal(100);
        connManager.setDefaultMaxPerRoute(100);

//        final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
//        credentialsProvider.setCredentials(AuthScope.ANY,
//                new UsernamePasswordCredentials(elasticSearchProperties.getBasic().getUsername(),
//                        elasticSearchProperties.getBasic().getUserpass()));
        RestClientBuilder builder = RestClient.builder(httpHosts).setHttpClientConfigCallback(callback -> {
            callback.disableAuthCaching();
            return callback.setKeepAliveStrategy((response, context) -> {
                final HeaderElementIterator it = new BasicHeaderElementIterator(
                        response.headerIterator(HTTP.CONN_KEEP_ALIVE));
                while (it.hasNext()) {
                    final HeaderElement he = it.nextElement();
                    final String param = he.getName();
                    final String value = he.getValue();
                    if (value != null && param.equalsIgnoreCase("timeout")) {
                        try {
                            return Long.parseLong(value) * 1000;
                        } catch (final NumberFormatException ignore) {
                        }
                    }
                }
                return 10 * 1000;
            }).setConnectionManager(connManager);
        }).setRequestConfigCallback(requestConfigBuilder -> {
            // 连接目标url的连接超时时间（毫秒）
            requestConfigBuilder.setConnectTimeout(5000);
            // 请求获取数据的超时时间(即响应时间)，单位毫秒
            requestConfigBuilder.setSocketTimeout(40000);
            // 从connect Manager(连接池)获取Connection 超时时间，单位毫秒
            requestConfigBuilder.setConnectionRequestTimeout(1000);
            return requestConfigBuilder;
        });
        return new RestHighLevelClient(builder);
    }


}